//Linux C plus plus file
/************************************************************************
*
*
*       			FileName:workerThread.cpp
*
*         			Author.wzj
*
*       			Email:345517039@qq.com
*
*      	    		Create:2017-06-22 23:39:58
*
*      		    	Description:-
*
*
*************************************************************************/
#include "workerThread.h"

/**
 * @brief 工作线程的构造函数
 */
workerThread::workerThread(){

}


/**
 * @brief 工作线程的析构函数
 */
workerThread::~workerThread(){

}


/**
 * @brief 工作线程的初始化工作
 */
int workerThread::init(void *arg)
{
#define LOG_BUF_MAX 1024
	std::vector<int> *local = (std::vector<int> *)arg;
	int socket[2] = { 0 };
	int iRet = RET::FAIL;
	mem.init(THREAD_LEVEL);
	struct workerThread_ctx * ctx = NULL;
	iRet = confMgr::getConfValue(SYSTEM_CONFIG,&serconf,conf);
	if (iRet == RET::FAIL){
		fprintf(stderr, "[ ERROR ]WORKER THREAD GET CONFIG OPTION FAILURE!\n");
		return iRet;
	}
	uworkThreadCount = serconf.processThreadNum; //处理线程的个数;
	DEBUG_TRACE(2,("[ INFO ]WORKER THREAD NUM:%lu\n", uworkThreadCount));
	for(size_t i = 0; i < uworkThreadCount; i++){
		ctx = (struct workerThread_ctx *)mem.allocateMemFromMempool(sizeof(struct workerThread_ctx));
		ctx->io = new epollModel();
		ctx->ioType = EPOLL_MODEL;
		ctx->io->init();
		ctx->conn.init();
		ctx->pth.func = workerThread::workerThreadRun;
		if( ctx->logfd == 0){ //后面考虑用双向队列;
			ctx->logfd = local->at(i);
			DEBUG_TRACE(2,("[ INFO ] INDEX:%lu WORKER THREAD LOG FD:%d\n", i, ctx->logfd));
			ctx->io->addEvent(local->at(i));
			local->at(i)= -1;
		}
		iRet = socketpair(AF_LOCAL, SOCK_STREAM, 0, socket);
		if ( iRet == RET::FAIL ){
			fprintf(stderr, "[ ERROR ] INDEX:%lu WORKER THREAD CREATE SOCKET PAIR FAILURE!\n", i);
			return iRet;
		}
		ctx->logbuf=(char *)mem.allocateMemFromMempool(LOG_BUF_MAX);
		if (!ctx->logbuf){
			fprintf(stderr, "[ ERROR ]ALLOC MEM FROM MEMPOOL OCCUR ERROR!\n");
			return RET::FAIL;
		}
		ctx->bufSize = LOG_BUF_MAX;
		ctx->sockfd = socket[1];	
		local->push_back(socket[0]);
		ctx->pth.arg = (void *)ctx;
		ctx->bStopFlag = false;
		ctx->io->addEvent(ctx->sockfd);
		iRet = pthread_create(&(ctx->pth.id), &(ctx->pth.attr),ctx->pth.func, ctx->pth.arg);
		if ( iRet == RET::FAIL ){
			fprintf(stderr,"[ ERROR ]WORKER THREAD CREATE FAILURE!\n");
			return iRet;
		}
		LogManger::LogFormatAndSend(ctx->logfd, INIT, ctx->logbuf, "sus", "WORKER THREAD TID:", (unsigned int)ctx->pth.id, "CREATE SUCCESSFULLY");
		DEBUG_TRACE(2,("[ INFO ]WORKER THREAD TID: %lu CREATE SUCCESSFULLY!\n", ctx->pth.id));
		this->workqueue.push_back(ctx);
	}
	return RET::SUCCESS;
}

/*@brief:
 *
 */
int workerThread::workerThreadHandle(int count, struct workerThread_ctx * ctx){
	int iRet = RET::FAIL;
	if ( !count || ctx == NULL ){
		return RET::FAIL;
	}
	int connectfd = 0;
	int ioType = ctx->ioType;
	switch(ioType){
	case EPOLL_MODEL:
		{
			EPOLL_CTX *epollctx =(EPOLL_CTX *)(ctx->io->getIoCtx());	
			for(int i = 0; i < count; i++ ){
				if ( epollctx->events[i].data.fd == ctx->sockfd && epollctx->events[i].events & EPOLLIN ){
					//read new connection from sockfd;
					iRet = read(ctx->sockfd, &connectfd, sizeof(connectfd));
					if ( iRet <= 0 ){
						return RET::FAIL;
					}
					//init this conenction;
					iRet = ctx->conn.initConnection(connectfd);
					if ( iRet == RET::FAIL ){
						DEBUG_TRACE(2,("[ ERROR ]WORKER THERAD TID %lu INIT A CONNECTION FAILURE!\n", ctx->pth.id));
						return iRet;
					}
					LogManger::LogFormatAndSend(ctx->logfd, NETWORKCONNECTION, ctx->logbuf, "susds", "WORKER THREAD TID", (unsigned int)ctx->pth.id, "GET A CONNECTION:", connectfd, "FROM LISTEN THREAD");
					DEBUG_TRACE(2,("[ CONNECTION ]WORKER THREAD TID: %lu GET A CONNECTION: %d FROM LISTEN THREAD\n", ctx->pth.id, connectfd));
				}else{

				}
			}
		}
		break;
	case SELECT_MODEL:
		{

		}
		break;
	case POLL_MODEL:
		{

		}
		break;
	default:
		{
			fprintf(stderr,"[ ERROR ]UNKNOWN IO MODEL TYPE SUPPORTED!\n");
			return RET::FAIL;
		}
	}
	return RET::SUCCESS;
}


/**
 * @brief 工作线程的回调函数
 *
 * @param arg
 *
 * @return 
 */
void *workerThread::workerThreadRun(void *arg)
{
	int iRet = RET::FAIL;
	prctl(PR_SET_NAME, "worker_thread");
	struct workerThread_ctx * worker_ctx = ( struct workerThread_ctx *)arg;
	if ( NULL == worker_ctx )
	{
		return NULL;
	}
	while (!worker_ctx->bStopFlag){
	
		iRet = worker_ctx->io->waitForEvent();
		if ( iRet == RET::FAIL )
		{
			return NULL;
		}else{
			iRet=workerThread::workerThreadHandle(iRet, worker_ctx);
			if ( iRet == RET::FAIL){
				worker_ctx->bStopFlag=1;
				DEBUG_TRACE(2,("[ ERROR ]WORKER THREAD EXIT ABNORMALLY!\n"));
			}
		}
	}
	return worker_ctx;
}



































